package cn.itcast.flink.source;

import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.FileProcessingMode;

/**
 * Author itcast
 * Date 2022/1/11 15:08
 * Desc 用于读取 File 文件中的数据
 */
public class FileSourceDemo {
    public static void main(String[] args) throws Exception {
        //1.获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度及其参数
        env.setParallelism(1);
        //3.读取文件
        //3.1 readTextFile 批处理，读完就结束
        DataStreamSource<String> source = env.readTextFile("D:\\project\\flinkbase27\\data\\words.txt");
        //3.2 readFile 两个参数 read once , 实时定时监控
        /*DataStreamSource<String> textSource = env.readFile(
                new TextInputFormat(null),
                "D:\\project\\flinkbase27\\data\\words.txt",
                FileProcessingMode.PROCESS_ONCE,
                5000
        );*/
        //读取文件 ， 流处理
        DataStreamSource<String> textSource = env.readFile(
                new TextInputFormat(null),
                "D:\\project\\flinkbase27\\data\\words.txt",
                FileProcessingMode.PROCESS_CONTINUOUSLY,
                5000
        );
        //4.打印输出
        textSource.printToErr();
        //5.执行流环境
        env.execute();
    }
}
